package cn.blnp.net.flowable.boot.ext.manager;

import cn.blnp.net.flowable.boot.bean.dto.CusJobMessage;
import cn.blnp.net.flowable.boot.utils.SpringContextUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.flowable.common.engine.api.FlowableException;
import org.flowable.engine.ProcessEngineConfiguration;
import org.flowable.engine.impl.history.async.HistoryJsonConstants;
import org.flowable.job.api.JobInfo;
import org.flowable.job.service.impl.asyncexecutor.message.AbstractMessageBasedJobManager;
import org.flowable.job.service.impl.history.async.transformer.HistoryJsonTransformer;
import org.flowable.job.service.impl.history.async.util.AsyncHistoryJsonUtil;
import org.flowable.job.service.impl.persistence.entity.HistoryJobEntityImpl;
import org.flowable.job.service.impl.persistence.entity.JobEntity;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * <h3>任务消息发送助手</h3>
 *
 * @author <a href="mailto:blnp.yibin@qq.com">lyb</a>
 * @version 1.0
 * @since 2025/6/23 16:16
 */
@Slf4j
public class RocketMqMessageBasedJobManager extends AbstractMessageBasedJobManager {

    public final static String JOB_TOPIC = "job_topic";
    public static final String JOB_CONSUMER_GROUP = "job_consumer_group";
    public final static String JOB_HISTORY_TOPIC = "job_history_topic";
    public final static String JOB_HISTORY_CONSUMER_GROUP = "job_history_group";

    @Override
    protected void sendMessage(JobInfo job) {
        RocketMQTemplate rocketMQTemplate = SpringContextUtils.getBean(RocketMQTemplate.class);
        if (job instanceof JobEntity) {
            JobEntity jobEntity = (JobEntity) job;
            //采用同步顺序推送
            rocketMQTemplate.syncSendOrderly(JOB_TOPIC,
                    //自定义推送消息对象
                    new CusJobMessage(jobEntity.getProcessInstanceId(), jobEntity.getId()),
                    jobEntity.getProcessInstanceId());
        }
        if (job instanceof HistoryJobEntityImpl) {
            HistoryJobEntityImpl historyJob = (HistoryJobEntityImpl) job;
            String historicalData = historyJob.getAdvancedJobHandlerConfiguration();
            ProcessEngineConfiguration configuration = (ProcessEngineConfiguration)SpringContextUtils.getBean("processEngineConfiguration");
            ObjectMapper objectMapper = configuration.getObjectMapper();
            JsonNode historyNode;
            String processInstanceId = "";
            try {
                historyNode = objectMapper.readTree(historicalData);
                if (historyNode instanceof ArrayNode) {
                    ArrayNode arrayNode = (ArrayNode) historyNode;
                    for (JsonNode jsonNode : arrayNode) {
                        processInstanceId = getProcessInstanceId(jsonNode);
                        if (StringUtils.isNotEmpty(processInstanceId)) {
                            break;
                        }
                    }
                } else {
                    processInstanceId = getProcessInstanceId(historyNode);
                }
            } catch (Exception e) {
                throw new FlowableException("Could not deserialize async " +
                        "history json for job (id=" + job.getId() + ")", e);
            }
            log.info("Send history data to MQ: {}", historicalData);
            CusJobMessage jobMessage = new CusJobMessage(processInstanceId, historyJob.getId());
            rocketMQTemplate.syncSendOrderly(JOB_HISTORY_TOPIC, jobMessage, processInstanceId);
        }
    }

    /**
     * <p><b>用途:解析获取流程实例ID<b></p>
     * @author <a href="mailto:blnp.yibin@qq.com">lyb</a>
     * @since 9:01 2025/6/24
     * @params [jsonNode]
     * @param jsonNode
     * @return java.lang.String
     **/
    private String getProcessInstanceId(JsonNode jsonNode) {
        ObjectNode jsonData = (ObjectNode) jsonNode
                .get(HistoryJsonTransformer.FIELD_NAME_DATA);
        String processInstanceId = AsyncHistoryJsonUtil.getStringFromJson(jsonData,
                HistoryJsonConstants.PROCESS_INSTANCE_ID);
        return processInstanceId == null ? "" : processInstanceId;
    }
}
